package com.smsc.headend.task.engine.controller;


import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.google.common.collect.Comparators;
import com.google.common.collect.Lists;
import com.smsc.headend.common.constant.Constant;
import com.smsc.headend.common.utils.PageUtils;
import com.smsc.headend.common.utils.R;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.common.AppName;
import com.smsc.headend.module.data.dto.LpDdsDTO;
import com.smsc.headend.module.data.dto.LpQueryDTO;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.data.entity.ScheduleReadingEntity;
import com.smsc.headend.module.report.entity.LogReportHis;
import com.smsc.headend.task.engine.message.DelayQueueMessage;
import com.smsc.headend.task.engine.po.ScheduleReadingDataBo;
import com.smsc.headend.task.engine.po.ScheduleReadingPageBo;
import com.smsc.headend.task.engine.service.DataConvertService;
import com.smsc.headend.task.engine.service.DataProcessService;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.report.GenReportService;
import com.smsc.headend.task.engine.service.task.LogReportHisService;
import com.smsc.headend.task.engine.utils.ExcelUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.i18n.LocaleContextHolder;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.scheduling.annotation.Async;
import org.springframework.web.bind.annotation.*;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.File;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;


@RestController
@Slf4j
@RequestMapping("lp")
public class RdMeterReadsLpController {

    @Autowired
    private DataProcessService dataProcessService;

    @Autowired
    DataConvertService dataConvertService;

    @Autowired
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    @Autowired
    FeignAssetManagementService assetManagementService;

    @Autowired
    LogReportHisService logReportHisService;

    @Autowired
    GenReportService genReportService;

    @PostMapping("mdmDds")
    public R taskProcessingInfo(@RequestBody LpDdsDTO lpDdsDto) {
        return dataProcessService.createExportLoadProfileTask(lpDdsDto);
    }

    @RequestMapping("queryLpData")
    public R lpDataQuery(Long meterId, Long lpDataItemId, Long startTv, Long endTv, Integer page, Integer pageSize) {
        if (page < 0) {
            return R.error("Illegal page");
        }

        if (pageSize > 2000 || pageSize <= 0) {
            return R.error("Page OverSize");
        }
        IPage<RdMeterReadsLp> lpData = dataProcessService.pageRdMeterReadsLp(lpDataItemId, meterId, startTv, endTv, page, pageSize);
        return R.data(lpData);
    }

    @RequestMapping("queryLpDataPivot")
    public R lpDataPivot(@RequestParam("meterId") Long meterId, @RequestParam("lpDataItemId") Long lpDataItemId
            , @RequestParam(value = "subDataItemIds", required = false) Long[] subDataItemIds
            , @RequestParam("startTv") Long startTv, @RequestParam("endTv") Long endTv, @RequestParam("page") Integer page
            , @RequestParam("pageSize") Integer pageSize, @RequestParam(required = false, value = "locale") String locale
            , @RequestParam(required = false, value = "CTVTConvert") Integer ctVtConvert) {
        if (page < 0) {
            return R.error("Illegal page");
        }
        if (pageSize > 2000 || pageSize <= 0) {
            return R.error("Page OverSize");
        }
        Date startDate = new Date(startTv * 1000);
        Date endDate = new Date(endTv * 1000);
        if (startDate.after(endDate)) {
            return R.error("end date must after start date");
        }
        if (!DateUtils.addMonths(startDate, 1).after(endDate)) {
            return R.error("End date after a month than start date");
        }
        ScheduleReadingPageBo scheduleReadingPageBo = dataProcessService.pagePivotRdMeterReadsLp(lpDataItemId, meterId, startTv, endTv, page, pageSize, subDataItemIds);
        PageUtils pageUtils = scheduleReadingPageBo.getPageUtils();
        Set<Long> dataItemIds = scheduleReadingPageBo.getDataItemIds();
        Map<String, Object> resultMap = new HashMap<>();

        List<ScheduleReadingEntity> dataList = (List<ScheduleReadingEntity>) pageUtils.getList();
        if (1 == ctVtConvert) {
            dataList = dataConvertService.convertCTVT(dataList);
        }
        resultMap.put("list", dataList);
        resultMap.put("totalRow", pageUtils.getTotalRow());
        resultMap.put("pageSize", pageUtils.getPageSize());
        resultMap.put("pageNumber", pageUtils.getPageNumber());
        resultMap.put("totalPage", pageUtils.getTotalPage());
        if (CollectionUtil.isNotEmpty(dataItemIds)) {
            Map<String, String> dataItemMap = dataConvertService.getSubDataItemIdName(Lists.newArrayList(dataItemIds), locale == null ? "en" : locale, Boolean.TRUE, AppName.GUI.toString());
            resultMap.put("columnNames", dataItemMap);
        }
        return R.data(resultMap);
    }

    @PostMapping("/export")
    public void export(HttpServletRequest request, HttpServletResponse response, @RequestBody Map<String, Object> params) throws IOException {
        log.info("export lp records, params:{}", JSONUtil.toJsonStr(params));
        Map criteria = (Map) params.get("criteria");
        ScheduleReadingDataBo scheduleReadingDataBo = getScheduleReadingDataBo(params, criteria);
        assetManagementService.getPropertyValue("");
        List<ScheduleReadingEntity> collect = getCollect(scheduleReadingDataBo, params);
        String[] titles = getTitles(scheduleReadingDataBo, params);
        String[] keys = getkeyList(scheduleReadingDataBo, params);
        String type = MapUtil.getStr(criteria, "type");
        ExcelUtils.write(response, "scheduleReadingRecords", collect, titles, keys, type);
    }


    @PostMapping("/report/generate")
    public R reportGenerate(@RequestBody Map<String, Object> params)  {
        log.info("export lp records, params:{}", JSONUtil.toJsonStr(params));
        Long userId = Convert.toLong(params.get("userId"));
        LogReportHis logReportHis = logReportHisService.saveReportLog(null, userId, LogReportHis.ReportType.SCHEDULE_READING.getCode(),SystemClock.now() / 1000, LogReportHis.ReportHisStatus.PROCESSING.getCode(),
                null, null, null, null);
        genReportService.generateReport(params, logReportHis);
        return R.ok();
    }

    private ScheduleReadingDataBo getScheduleReadingDataBo(Map<String, Object> params, Map criteria) {
        List<Long> meterIds = (List<Long>) ((Map) params.get("criteria")).get("meterIds");
        List<Integer> dcuIds = (List<Integer>) ((Map) params.get("criteria")).get("dcuIds");
        if (meterIds.isEmpty() && !dcuIds.isEmpty()) {
            if(meterIds == null) {
                meterIds = CollUtil.newArrayList();
            }
            for (Integer dcuId : dcuIds) {
                List<Meter> meterList = assetManagementService.getMetersByDcuId(Convert.toLong(dcuId));
                meterIds.addAll(meterList.stream().map(Meter::getMeterId).collect(Collectors.toList()));
            }
        }
        List<Long> timeRangeList = (List) criteria.get("timeRange");
        List<Long> subDataItemIds = (List) criteria.get("subDataItemArr");
        Long lpDataItemId = MapUtil.getLong(criteria, "dataGroup");
        Long startTime = timeRangeList.get(0) / 1000;
        Long endTime = timeRangeList.get(1) / 1000;
        Long[] paramMeters = Convert.toLongArray(meterIds);
        ScheduleReadingDataBo scheduleReadingDataBo = dataProcessService.lpDataPivotByOffsetAndLimit(paramMeters, lpDataItemId, Convert.toLongArray(subDataItemIds), startTime, endTime, 0, Constant.EXPORT_LIMIT);

        return scheduleReadingDataBo;
    }

    private String[] getTitles(ScheduleReadingDataBo scheduleReadingDataBo, Map<String, Object> params) {
        String locale = MapUtil.getStr(params, "locale");
        List<String> titleList = CollUtil.toList(dataConvertService.getI18NStr("customerSerialNumber", locale), dataConvertService.getI18NStr("time", locale));//CollUtil.toList("Customer Serial Number", "Time");
        Map<String, String> subDataItemIdName = dataConvertService.getSubDataItemIdName(Lists.newArrayList(scheduleReadingDataBo.getDataItemIds()), locale, Boolean.TRUE, AppName.GUI.toString());
        subDataItemIdName.entrySet().forEach(e -> {
            titleList.add(e.getValue());
        });
        titleList.add(dataConvertService.getI18NStr("profileStatus", locale));
        titleList.add(dataConvertService.getI18NStr("source", locale));//titleList.add("Source");
        titleList.add(dataConvertService.getI18NStr("updateTime", locale));//titleList.add("Update Time");
        return ArrayUtil.toArray(titleList, String.class);
    }

    private String[] getkeyList(ScheduleReadingDataBo scheduleReadingDataBo, Map<String, Object> params) {
        String locale = MapUtil.getStr(params, "locale");
        List<String> keyList = CollUtil.toList("customerSn", "tv");
        Map<String, String> subDataItemIdName = dataConvertService.getSubDataItemIdName(Lists.newArrayList(scheduleReadingDataBo.getDataItemIds()), locale, Boolean.TRUE, AppName.GUI.toString());
        subDataItemIdName.entrySet().forEach(e -> {
            keyList.add(e.getKey());
        });
        keyList.add("validity");
        keyList.add("src");
        keyList.add("tvUpdate");
        return ArrayUtil.toArray(keyList, String.class);
    }

    private List<ScheduleReadingEntity> getCollect(ScheduleReadingDataBo scheduleReadingDataBo, Map<String, Object> params) {
        Map criteria = (Map) params.get("criteria");
        Integer calculateRatio = MapUtil.getInt(criteria, "calculateRatio");
        List<ScheduleReadingEntity> dataList = scheduleReadingDataBo.getScheduleReadingEntityList();

        List<ScheduleReadingEntity> collect = dataList.stream().sorted((d1, d2) -> {
            String customerSn1 = MapUtil.getStr(d1, "customerSn");
            String customerSn2 = MapUtil.getStr(d2, "customerSn");
            int i = customerSn1.compareTo(customerSn2);
            if (i != 0) {
                return i;
            }
            String tv1 = MapUtil.getStr(d1, "tv");
            String tv2 = MapUtil.getStr(d2, "tv");
            return tv1.compareTo(tv2) * -1;
        }).collect(Collectors.toList());
        if (1 == calculateRatio) {
            collect = dataConvertService.convertCTVT(collect);
        }
        return collect;
    }

    @PostMapping("queryLpDataPivotOffsetMode")
    public R lpDataPivotByOffsetAndLimit(@RequestBody Map<String, Object> params) {
        List<Integer> meterIds = (List<Integer>) params.get("meterIds");
        Long[] mIds = meterIds.stream().map(id -> Convert.toLong(id)).toArray(Long[]::new);
        Long lpDataItemId = MapUtil.getLong(params, "lpDataItemId");
        List<Long> subDataItemIds = (List<Long>) params.get("subDataItemIds");
        Long[] sIds = subDataItemIds.stream().map(id -> Convert.toLong(id)).toArray(Long[]::new);
        subDataItemIds.toArray(sIds);
        Long startTv = MapUtil.getLong(params, "startTv");
        Long endTv = MapUtil.getLong(params, "endTv");
        Integer offset = MapUtil.getInt(params, "offset");
        Integer limit = MapUtil.getInt(params, "limit");
        String locale = MapUtil.getStr(params, "locale");
        Integer ctVtConvert = MapUtil.getInt(params, "CTVTConvert");

        if (limit > 20000) {
            return R.error("Limit OverSize");
        }
        Date startDate = new Date(startTv * 1000);
        Date endDate = new Date(endTv * 1000);
        if (startDate.after(endDate)) {
            return R.error("end date must after start date");
        }
        if (!DateUtils.addMonths(startDate, 1).after(endDate)) {
            return R.error("End date after a month than start date");
        }
        ScheduleReadingDataBo scheduleReadingDataBo = dataProcessService.lpDataPivotByOffsetAndLimit(mIds, lpDataItemId, sIds, startTv, endTv, offset, limit);
        Map<String, Object> map = new HashMap<>();
        Map<String, String> dataItemMap = dataConvertService.getSubDataItemIdName(Lists.newArrayList(scheduleReadingDataBo.getDataItemIds()), locale, Boolean.TRUE, AppName.GUI.toString());
        List<ScheduleReadingEntity> dataList = scheduleReadingDataBo.getScheduleReadingEntityList();
        if (1 == ctVtConvert) {
            dataList = dataConvertService.convertCTVT(dataList);
        }
        map.put("list", dataList);
        map.put("columnNames", dataItemMap);
        return R.data(map);
    }


    @RequestMapping("queryLpDataByRange")
    public R lpDataQueryByRange(Long[] meterIds, Long lpDataItemId, Long startTv, Long endTv, Integer page, Integer pageSize) {

        if (page < 0) {
            return R.error("Illegal page");
        }

        if (pageSize > 2000 || pageSize <= 0) {
            return R.error("Page OverSize");
        }
        IPage<RdMeterReadsLp> lpData = dataProcessService.pageRdMeterReadsLpByRange(lpDataItemId, Lists.newArrayList(meterIds), startTv, endTv, page, pageSize);
        return R.data(lpData);
    }

    @RequestMapping("countMetersDataByTv")
    public Integer countMetersDataByTv(Long[] meterIds, Long lpDataItemId, Long startTv, Long endTv) {
        Long start = SystemClock.now();
        Integer integer = dataProcessService.countByTvs(lpDataItemId, Lists.newArrayList(meterIds), startTv, endTv);
        log.info("count: cost:{}ms", SystemClock.now() - start);
        return integer;
    }

    @RequestMapping("queryLpDataOnTv")
    public R readElectricMeterFrozenData(@RequestBody LpQueryDTO lpQueryDTO) {
        List<RdMeterReadsLp> readsLps = dataProcessService.queryElectricMeterData(lpQueryDTO);
        if (CollectionUtils.isEmpty(readsLps)) {
            return R.ok("no data");
        }
        return R.data(readsLps);
    }

    @RequestMapping("queryLpDataByDailyOnTv")
    public R readElectricMeterFrozenDataByDaily(@RequestBody LpQueryDTO lpQueryDTO) {
        List<RdMeterReadsLp> readsLps = dataProcessService.queryElectricMeterDataByDaily(lpQueryDTO);
        if (CollectionUtils.isEmpty(readsLps)) {
            return R.ok("no data");
        }
        return R.data(readsLps);
    }

//    @RequestMapping("testListenerSTSP")
//    public R testListenerSTSP() throws InterruptedException {
//        MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry.getListenerContainer(DelayQueueMessage.DELAY_HOUR_4);
//        boolean containerPaused = listenerContainer.isContainerPaused();
//        log.info("container status" + " {}", containerPaused);
//        listenerContainer.resume();
//        log.info("resume");
//        log.info("container status" + " {}", containerPaused);
//        return R.ok();
//    }
//
//    @RequestMapping("testListenerStatus")
//    public R testListenerStatus() throws InterruptedException {
//        MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry.getListenerContainer(DelayQueueMessage.DELAY_HOUR_4);
//        boolean containerPaused = listenerContainer.isContainerPaused();
//        log.info("container status" + " {}", containerPaused);
//        return R.ok();
//    }
//
//
//    @RequestMapping("testListenerPause")
//    public R testListenerResume() throws InterruptedException {
//        MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry.getListenerContainer(DelayQueueMessage.DELAY_HOUR_4);
//        boolean containerPaused = listenerContainer.isContainerPaused();
//        log.info("container status" + " {}", containerPaused);
//        listenerContainer.pause();
//        log.info("pause");
//        return R.ok();
//    }

}
